package com.atguigu.flink.datastramapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/2/25
 *
 *  无keyBy，不聚合。
 *
 *      flink提供了 sum,max,min
 *                      输出的记录，除了分组字段和聚合字段，其他字段取组内第一条数据的属性。
 *
 *                  maxBy:  取记录最大的所有信息
 *                  minBy:   取记录最小的所有信息
 */
public class Demo5_Agg
{
    public static void main(String[] args) {

      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        //求每种传感器的水位和
        ds.keyBy(WaterSensor::getId)
          //.max("vc")
          // 做max运算，除了分组列和聚合列，其他列取最大那条数据的。如果出现并列的，取第一条最大的。
          //.maxBy("vc")
          // 做max运算，除了分组列和聚合列，其他列取最大那条数据的。如果出现并列的，取第最后一条最大的。
          .maxBy("vc",false)
          .print();




        try {
                          env.execute();
                      } catch (Exception e) {
                          e.printStackTrace();
                      }


    }
}
